/**
 * $Id: RunnerMgr.java,v 1.29 2006/09/30 05:32:47 martin Exp $
 *
 * Copyright 2006 Skyinfo, Inc. All Rights Reserved.
 */
package com.haoxi.gateway.runner;

import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.haoxi.gateway.common.GatewayInfo;
import com.haoxi.gateway.common.Queue;
import com.haoxi.gateway.common.SYSConst;
import com.haoxi.gateway.connector.Connector;
import com.haoxi.gateway.messageconvertor.MessageConvertor;
import com.haoxi.gateway.runner.datadeal.DealOverDateDataMgr;
import com.haoxi.gateway.runner.datadeal.DealOverDateNeedReportMTThread;
import com.haoxi.gateway.runner.datadeal.DealOverDateNeedRespMTThread;
import com.haoxi.gateway.runner.process.DealReportHistoryThread;
import com.haoxi.gateway.runner.process.DispathcerThread;
import com.haoxi.gateway.runner.process.ReportReciveThread;
import com.haoxi.gateway.runner.process.UserMOThread;
import com.haoxi.gateway.runner.specialmobile.SpecialMobileSyncThread;

// import com.haoxi.gateway.runner.synchdata.SyncBussinessDataThread;

/**
 * 加载连接器和消息转换配置器文件（可以采用依赖注入原则）、队列定义初始化、业务线程启动和调度
 * 
 * Revision History
 * 
 * 2006-5-25,star.Zhao,created it
 */
public class RunnerMgr {
    private static final Log log = LogFactory.getLog(RunnerMgr.class);

    private static RunnerMgr instance;
    private GatewayInfo gatewayInfo;

    private Map systemConf;

    /**
     * 保留用于将来扩展Quene数据保护的开关
     */
    private int memorydata_protect;

    /**
     * send mt thread
     */
    private AbsDaemon submitDeliverRespDeamon;

    /**
     * send mt thread
     */
    private AbsDaemon submitStatusDeamon;

    /**
     * queue manager
     */
    private QueueMgr queueMgr = QueueMgr.getInstance();

    /**
     * recive queue
     */
    private Queue reciveQueue = queueMgr.getReciveQueue();
    /**
     * user MO queue
     */
    private Queue userMOQueue = queueMgr.getUserMOQueue();

    /**
     * submit status queue
     */
    private Queue submitStatusQueue = queueMgr.getSubmitStatusQueue();
    /**
     * submit report queue
     */
    private Queue submitReportQueue = queueMgr.getSubmitReportQueue();
    /**
     * submit queue
     */
    private Queue submitMTQueue = queueMgr.getSubmitMTQueue();

    /**
     * submit queue
     */
    private Queue submitRespQueue = queueMgr.getSubmitRespQueue();

    /**
     * save the not a submit status msg objects by sequence
     */
    private Map submitRespMap = Collections.synchronizedMap(new HashMap());

    /**
     * recive deliver Connection Deamon
     */
    private AbsConnectorDaemon reciverDeliverConnDeamon;

    /**
     * Submit mt Connection Deamon
     */
    private AbsConnectorDaemon submitMTConnDeamon;

    /**
     * dispatch thread
     */
    private IRun dispatchThread;

    /**
     * process get userMo thread
     */
    private IRun userMOThread;

    /**
     * process get submit Report thread
     */
    private IRun submitReportThread;

    public synchronized static RunnerMgr getInstance() {
        if (instance == null) {
            instance = new RunnerMgr();
        }
        return instance;
    }

    /**
     * 启动ClientConnector
     * 
     * @param allConnList
     * @return
     */
    private boolean startupCilentConnector(List allConnList) {
        boolean bOK = false;
        log.debug("RunnerMgr start startup all client connectors......:"+allConnList.size());
        if (null == allConnList || allConnList.size() == 0) {
            log.error("The system Connector config wrong, please check it!");
            return bOK;
        }
        Iterator iterator = allConnList.iterator();
        while (iterator.hasNext()) {
            Connector connector = (Connector) iterator.next();
            if (!connector.openConnect()) {
                log.error("RunnerMgr startup the client connector["
                        + connector.getConnName() + "] unsuccessfully!");
                return bOK;
            }
        }
        bOK = true;
        log.debug("RunnerMgr startup all client connectors successfully!");
        return bOK;
    }

    /**
     * 启动ServerConnector
     * 
     * @param serConnList
     * @return
     */
    private boolean startupServerConnector(List serConnList) {
        boolean bOK = false;
        log.debug("RunnerMgr start startup all server connectors......:"+serConnList.size());
        if (null != serConnList && serConnList.size() > 0) {
            Iterator iterator = serConnList.iterator();
            while (iterator.hasNext()) {
                Connector connector = (Connector) iterator.next();
                if (!connector.openConnect()) {
                    log.error("RunnerMgr startup the server connector["
                            + connector.getConnName() + "] unsuccessfully!");
                    return bOK;
                }
            }
        }
        bOK = true;
        log.debug("RunnerMgr startup all server connectors successfully!");
        return bOK;
    }

    public synchronized boolean startRunner() {
        
        log.debug("RunnerMgr startRunner starting ...");
        boolean bOK = false;
        if (gatewayInfo == null) {
            log.error("init gatewayInfo failure!!!");
            return bOK;
        }
        // get the right convertor for the version cmpp message
        MessageConvertor convertor = gatewayInfo.getConvertor();

        // 加载MTRespMap同步表中的数据和MessageConvertor中SequenceID的初始值
        if (!DealOverDateDataMgr.getInstance().initSystemDeal()) {
            return bOK;
        }
        
        // startup ServerConnector
        if (!this.startupServerConnector(gatewayInfo.getServerConnectors())) {
            return bOK;
        }

        // startup ClientConnector
        if (!this.startupCilentConnector(gatewayInfo.getClientConnectors())) {
            return bOK;
        }

        // start receive deliver Connection Deamon
        reciverDeliverConnDeamon = new ReciverDeliverConnDeamon(gatewayInfo.getClientConnectors(), reciveQueue);
        reciverDeliverConnDeamon.start();

        //屏蔽 by dean
        // start dispatch thread
        dispatchThread = new DispathcerThread(reciveQueue, convertor);
        Thread t_dispatcherThread = new Thread(dispatchThread);
        t_dispatcherThread.start();
        //end 屏蔽
        
        // start save mo thread
        userMOThread = new UserMOThread(userMOQueue);
        Thread t_userMOThread = new Thread(userMOThread);
        t_userMOThread.start();

        // start getMT and submitMT thread ConnDeamon
        
        submitMTConnDeamon = new SubmitAndGetMTConnDeamon(submitMTQueue,reciveQueue,
                gatewayInfo.getCanSubmitMTConnectors(), submitRespMap,
                convertor);
        submitMTConnDeamon.start();
        log.debug("submitMTQueue:"+submitMTQueue.getLength()+"reciveQueue:"+reciveQueue.getLength()+"submitRespMap:"+submitRespMap.size());

        // start submit deliver Resp Thread
        submitDeliverRespDeamon = new SubmitDeliverRespDeamon(submitRespQueue);
        submitDeliverRespDeamon.setMax_thread_count(2);
        Thread t_submitDeliverRespDeamon = new Thread(submitDeliverRespDeamon);
        t_submitDeliverRespDeamon.start();

        // start save submitStatus thread
        submitStatusDeamon = new SubmitStatusDeamon(submitStatusQueue,
                submitRespMap);
        Thread t_submitStatusDeamon = new Thread(submitStatusDeamon);
        t_submitStatusDeamon.start();

        // start deal submitReport two threads
        submitReportThread = new ReportReciveThread(submitReportQueue);
        Thread t_submitReportThread = new Thread(submitReportThread);
        t_submitReportThread.start();
        // 定时匹配Report临时记录表和等待Report的MT临时表数据
//        Timer timer=new Timer(true);
//        TimerTask dealReportHistory=new DealReportHistoryThread();
//        timer.schedule(dealReportHistory,new Date(),SYSConst.CHECK_REPORT_MT_DATA_TIME);
//        Runnable dealReportHistory = new DealReportHistoryThread(SYSConst.CHECK_REPORT_MT_DATA_TIME);
//        new Thread(dealReportHistory).start();

        // 定时检测MT_RESPMap和Report相应的临时表中超时数据
//        Runnable dealOverDateData = new DealOverDateNeedRespMTThread(
//                submitRespMap, SYSConst.CHECK_OVERDATE_RESPDATA_TIME);
//        new Thread(dealOverDateData).start();
//        Runnable dealOverDateReportData = new DealOverDateNeedReportMTThread(
//                SYSConst.CHECK_OVERDATE_REPORTDATA_TIME);
//        new Thread(dealOverDateReportData).start();

        //屏蔽 by dean
//        // 定时监测敏感号和黑名单
//        Runnable specialMobileSyncThread = new SpecialMobileSyncThread(
//                SYSConst.CHECK_SPECIALMOBILE_CHG_TIME);
//        new Thread(specialMobileSyncThread).start();
        //end 屏蔽
        // 定时同步敏感号和黑名单等
        // Runnable blackSyncThread = new
        // SyncBussinessDataThread(SYSConst.CHECK_SPECIALMOBILE_CHG_TIME,
        // SYSConst.SYNC_BLACKMOBILE_TYPE, gatewayInfo.getGatewayCode(), "ALL");
        // new Thread ( blackSyncThread ).start ();
        // Runnable senseSyncThread = new
        // SyncBussinessDataThread(SYSConst.CHECK_SPECIALMOBILE_CHG_TIME,
        // SYSConst.SYNC_SENSEMOBILE_TYPE, gatewayInfo.getGatewayCode(),
        // "NONE");
        // new Thread ( senseSyncThread ).start ();
        // Runnable blackLongNumSyncThread = new
        // SyncBussinessDataThread(SYSConst.CHECK_SPECIALMOBILE_CHG_TIME,
        // SYSConst.SYNC_BLACKLONGNUM_TYPE, gatewayInfo.getGatewayCode(),
        // "NONE");
        // new Thread ( blackLongNumSyncThread ).start ();

        //屏蔽 by dean
//        // 定时监测系统配置控制参数
//        Runnable sysControlConfSyncThread = new SysControlConfSyncThread(
//                SYSConst.CHECK_CONTROL_SYSCONF_CHG_TIME);
//        new Thread(sysControlConfSyncThread).start();
        //end 屏蔽 
        
        bOK = true;
        log.debug("RunnerMgr startRunner OK !!!");
        return bOK;
    }

    public synchronized boolean stopRunner() {
        log.debug("RunnerMgr stopRunner ing ...");
        boolean bOK = false;

        return bOK;
    }

    /**
     * test method
     * 
     * @param argv
     * @return
     */
    public static void main(String[] argv) {
        RunnerMgr mgr = new RunnerMgr();
        mgr.startRunner();

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        mgr.stopRunner();
    }

    /**
     * @return Returns the memorydata_protect.
     */
    public int getMemorydata_protect() {
        return memorydata_protect;
    }

    /**
     * @param memorydata_protect The memorydata_protect to set.
     */
    public void setMemorydata_protect(int memorydata_protect) {
        this.memorydata_protect = memorydata_protect;
    }

    /**
     * @return Returns the systemConf.
     */
    public Map getSystemConf() {
        return systemConf;
    }

    /**
     * @param systemConf The systemConf to set.
     */
    public void setSystemConf(Map systemConf) {
        this.systemConf = systemConf;
    }

    /**
     * @param gatewayInfo The gatewayInfo to set.
     */
    public void setGatewayInfo(GatewayInfo gatewayInfo) {
        this.gatewayInfo = gatewayInfo;
    }

    /**
     * @return Returns the gatewayInfo.
     */
    public GatewayInfo getGatewayInfo() {
        return gatewayInfo;
    }
}
